package ltd.dujiabao.zookeeper.watcher;

import org.apache.log4j.Logger;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.CountDownLatch;

/**
 * @author : DuJiabao
 * @Project : zookeeper
 * @Time : 2021/5/3 14:10
 * @Desc :
 */
public class DistributedLock {
  private String ip = null;  // ip及端口
  private static Integer timeOut = 5000;  // 超时时间，毫秒为单位
  private final static CountDownLatch countDownLatch = new CountDownLatch(1);
  private final static Logger log = Logger.getLogger(DistributedLock.class);
  private ZooKeeper zooKeeper = null;
  private static final String LOCK_ROOT_PATH = "/Locks";
  private static final String LOCK_NODE_NAME = "Lock_";
  String lockPath; // 锁节点的路径

  public DistributedLock() {
  }

  public DistributedLock(String ip) {
    this(ip, timeOut);
  }

  public DistributedLock(String ip, Integer timeOut) {
    this.ip = ip;
    DistributedLock.timeOut = timeOut;
    try {
      initZK(ip, timeOut);
    } catch (IOException | InterruptedException e) {
      e.printStackTrace();
    }
  }

  private void initZK(String ip, Integer timeOut) throws IOException, InterruptedException {
    zooKeeper = new ZooKeeper(ip, timeOut, new Watcher() {
      @Override
      public void process(WatchedEvent watchedEvent) {
        if (watchedEvent.getState() == Event.KeeperState.SyncConnected) {
          log.info("连接成功");
          countDownLatch.countDown();
        }
      }
    });
    countDownLatch.await();
  }

  public void release() {
    try {
      zooKeeper.delete(lockPath, -1);  // 删除本节点
      zooKeeper.close();  // 关闭客户端
      log.info("已释放" + this.lockPath);
    } catch (InterruptedException | KeeperException e) {
      e.printStackTrace();
    }
  }

  /**
   * 需要锁，如果没有到自己，那就阻塞等
   */
  public void acquire() {
    try {
      createLock();  // 创建锁节点
      attemptLock();  // 祈求获取节点，没有的话就阻塞
    } catch (KeeperException | InterruptedException e) {
      e.printStackTrace();
    }
  }

  /**
   * 创建一个自己的号码
   */
  private void createLock() throws KeeperException, InterruptedException {
    if (zooKeeper.exists(LOCK_ROOT_PATH, true) == null) {
      zooKeeper.create(LOCK_ROOT_PATH, "0".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
      log.info("创建父节点: " + LOCK_ROOT_PATH + " 成功");
    }
    lockPath = zooKeeper.create(LOCK_ROOT_PATH + "/" + LOCK_NODE_NAME, "0".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
    log.info("创建节点成功：" + lockPath);
  }

  /**
   * 一个监视器，监视上一个节点的删除
   */
  Watcher watcher = new Watcher() {
    @Override
    public void process(WatchedEvent watchedEvent) {
      if (watchedEvent.getType() == Event.EventType.NodeDeleted) {
        synchronized (this) {
          notifyAll();
        }
      }
    }
  };

  /**
   * 获取锁。如果还没轮到自己，那就wait(),直到前一个节点的删除，唤醒自己
   */
  private void attemptLock() throws KeeperException, InterruptedException {
    List<String> children = zooKeeper.getChildren(LOCK_ROOT_PATH, null);
    Collections.sort(children);
    int index = children.indexOf(lockPath.substring(LOCK_ROOT_PATH.length() + 1));
    if (index == 0) {
      log.info("获取锁成功");
      return;
    }
    String lastPath = children.get(index - 1);  // 获取上一个节点的path
    // 添加一个监视前一个节点的监视器
    Stat stat = zooKeeper.exists(LOCK_ROOT_PATH + "/" + lastPath, watcher);
    if (stat == null) {  // 如果目前是空的，说明已经被删除了
      attemptLock();
      return;
    }
    synchronized (watcher) {
      watcher.wait();
    }
    attemptLock();
  }
}
